Exercise for Convolutional Neural Networks: PdM Classification
By Prof. Seungchul Lee
http://iai.postech.ac.kr/
Industrial AI Lab at POSTECH
http://iai.postech.ac.kr/
Industrial AI Lab at POSTECH
Table of Contents
1. Problem Definition¶
- Predictive Maintenance
- This predictive maintenance project focuses on the techniques used to predict when an in-service machine will fail, so that maintenance can be planned in advance.
- Overview of the RUL estimation strategy
- The remaining life of a test unit is estimated based on the actual life of a training unit that has the most similar degradation pattern.
- Data description
- Training data: It is the aircraft engine run-to-failure data.
- Testing data: It is the aircraft engine operating data without failure events recorded.
- Ground truth data: It contains the information of true remaining cycles for each engine in the testing data.
- Operational conditions
- Altitude (0 ~ 42,000 feet), Mach number (0 ~ 0.84) and throttle resolver angle (20 ~ 100)
- 21 sensors
- Different measurements related to the engine state at runtime.
2. Import Library¶
In [1]:
import pandas as pd
import numpy as np
import tensorflow as tf
import matplotlib.pyplot as plt
from sklearn import preprocessing
from sklearn.metrics import confusion_matrix
import seaborn as sn
3. Load Data¶
In [2]:
train_df = pd.read_csv('./data_files/PM_train.txt', sep=" ", header=None)
train_df.head()
Out[2]:
In [3]:
test_df = pd.read_csv('./data_files/PM_test.txt', sep=" ", header=None)
test_df.head()
Out[3]:
In [4]:
truth_df = pd.read_csv('./data_files/PM_truth.txt', sep=" ", header=None)
truth_df.head()
Out[4]:
3.1. Plot¶
In [5]:
def plot_sensor_data(data, purpose):
for j in data.keys().difference([0,1,26,27]).values:
max_cycle = data.groupby([0]).count()[j].max()
if purpose == 'train':
plt.figure(figsize = (20, 1.5))
for i in range(data[0].max()):
padding = max_cycle - data[data[0] == i][j].values.shape[0]
plt.plot(np.arange(padding, max_cycle), data[data[0] == i][j].values)
if j == 2 or j == 3 or j == 4:
plt.title(str(j-1) + 'th setting')
else:
plt.title(str(j-4) + 'th sensor')
plt.vlines(max_cycle, data[j].min(), data[j].max(), colors = 'r', linestyles = 'dashed')
plt.text(max_cycle+1, data[j].median(), "failure", fontsize=14)
plt.show()
if purpose == 'test':
plt.figure(figsize = (20, 1.5))
for i in range(data[0].max()):
plt.plot(data[data[0] == i][j].values)
if j == 2 or j == 3 or j == 4:
plt.title(str(j-1) + 'th setting')
else:
plt.title(str(j-4) + 'th sensor')
plt.text(max_cycle+1, data[j].median(), "on working", fontsize=14)
plt.show()
In [6]:
# plot train data
plot_sensor_data(train_df, 'train')
In [7]:
# plot test data
plot_sensor_data(test_df, 'test')
4. Pre-processing¶
4.1. Drop NaN column¶
In [8]:
train_df.drop(train_df.columns[[26, 27]], axis=1, inplace=True)
train_df.columns = ['id', 'cycle', 'setting1', 'setting2', 'setting3', 's1', 's2', 's3',
's4', 's5', 's6', 's7', 's8', 's9', 's10', 's11', 's12', 's13', 's14',
's15', 's16', 's17', 's18', 's19', 's20', 's21']
test_df.drop(test_df.columns[[26, 27]], axis=1, inplace=True)
test_df.columns = ['id', 'cycle', 'setting1', 'setting2', 'setting3', 's1', 's2', 's3',
's4', 's5', 's6', 's7', 's8', 's9', 's10', 's11', 's12', 's13', 's14',
's15', 's16', 's17', 's18', 's19', 's20', 's21']
truth_df.drop(truth_df.columns[[1]], axis=1, inplace=True)
4.2. Train Data¶
In [9]:
train_df.head()
Out[9]:
4.2.1. Labeling¶
In [10]:
train_df['RUL'] = train_df.groupby(['id'])['cycle'].transform(max) - train_df['cycle']
train_df.head()
Out[10]:
4.2.2. Normalize¶
In [11]:
cols_normalize = train_df.columns.difference(['id','cycle','RUL'])
cols_normalize
Out[11]:
In [12]:
# normalize
min_max_scaler = preprocessing.MinMaxScaler()
norm_train_df = pd.DataFrame(min_max_scaler.fit_transform(train_df[cols_normalize]),
columns=cols_normalize,
index=train_df.index)
norm_train_df.head()
Out[12]:
In [13]:
join_df = train_df[train_df.columns.difference(cols_normalize)].join(norm_train_df)
train_df = join_df.reindex(columns = train_df.columns)
train_df.head()
Out[13]:
4.3. Test Data¶
4.3.1. Labeling¶
In [14]:
# Get value of max cycle
rul = pd.DataFrame(test_df.groupby('id')['cycle'].max()).reset_index()
rul.columns = ['id', 'max']
rul.head()
Out[14]:
In [15]:
truth_df.columns = ['more']
truth_df['id'] = truth_df.index + 1
truth_df.head()
Out[15]:
In [16]:
truth_df['rtf'] = truth_df['more'] + rul['max']
truth_df.drop('more', axis = 1, inplace = True)
truth_df.head()
Out[16]:
In [17]:
# Get RUL
test_df = test_df.merge(truth_df)
test_df['RUL'] = test_df['rtf'] - test_df['cycle']
test_df.drop('rtf', axis = 1, inplace = True)
test_df.head()
Out[17]:
In [18]:
# Labeling
period = 30
train_df['label_bc'] = train_df['RUL'].apply(lambda x: 1 if x <= period else 0)
test_df['label_bc'] = test_df['RUL'].apply(lambda x: 1 if x <= period else 0)
train_df.head()
Out[18]:
4.3.2. Normalize¶
In [19]:
norm_test_df = pd.DataFrame(min_max_scaler.transform(test_df[cols_normalize]),
columns=cols_normalize,
index=test_df.index)
test_join_df = test_df[test_df.columns.difference(cols_normalize)].join(norm_test_df)
test_df = test_join_df.reindex(columns = test_df.columns)
test_df = test_df.reset_index(drop=True)
test_df.head()
Out[19]:
4.4. Data Generation¶
In [20]:
# function to generate input sequences
def gen_sequence(id_df, n_step, seq_cols):
df_zeros = pd.DataFrame(np.zeros((n_step - 1, id_df.shape[1])), columns = id_df.columns) # zero padding
id_df = df_zeros.append(id_df, ignore_index = True) # zero padding
data_array = id_df[seq_cols].values
num_elements = data_array.shape[0]
lstm_array = []
for start, stop in zip(range(0, num_elements - n_step), range(n_step, num_elements)):
lstm_array.append(data_array[start:stop, :])
return np.array(lstm_array)
# function to generate labels
def gen_label(id_df, n_step, seq_cols, label):
df_zeros=pd.DataFrame(np.zeros((n_step - 1, id_df.shape[1])), columns = id_df.columns) # zero padding
id_df=df_zeros.append(id_df, ignore_index = True) # zero padding
data_array = id_df[seq_cols].values
num_elements = data_array.shape[0]
y_label=[]
for start, stop in zip(range(0, num_elements - n_step), range(n_step, num_elements)):
y_label.append(id_df[label][stop])
return np.array(y_label)
In [21]:
# generate data
n_step = 50
n_sensor = 24
sensor_cols = ['s' + str(i) for i in range(1,22)]
x_cols = ['setting1', 'setting2', 'setting3']
x_cols.extend(sensor_cols)
X_train = np.concatenate(list(list(gen_sequence(train_df[train_df['id'] == id], n_step, x_cols)) for id in train_df['id'].unique()))
y_train = np.concatenate(list(list(gen_label(train_df[train_df['id'] == id], n_step, x_cols,'label_bc')) for id in train_df['id'].unique()))
X_test=np.concatenate(list(list(gen_sequence(test_df[test_df['id']==id], n_step, x_cols)) for id in test_df['id'].unique()))
y_test=np.concatenate(list(list(gen_label(test_df[test_df['id']==id], 50, x_cols,'label_bc')) for id in test_df['id'].unique()))
y_train = y_train[:,np.newaxis]
y_test = y_test[:,np.newaxis]
print(X_train.shape)
print(y_train.shape)
print(X_test.shape)
print(y_test.shape)
5. Build a Model¶
5.1. 1D Convolution¶
- Eventhough input is 2D ex) 50x24, output-shape is not 2D , but 1D Matrix
- What if we want to train N filters (N is number of filters)? then output shape is (stacked 1D) 2D = 1D x N matrix
In [22]:
# define the CNN structure and placeholder
input_length = n_step
input_ch = n_sensor
n_hidden = 32
n_output = 1
x = tf.placeholder(tf.float32, [None, input_length, input_ch])
y = tf.placeholder(tf.float32, [None, n_output])
In [23]:
# build a model
def net(x):
## First convolution layer
conv1 = tf.layers.conv1d(inputs = x,
filters = 8,
kernel_size = 7,
padding = 'SAME',
activation = tf.nn.relu)
maxp1 = tf.layers.max_pooling1d(inputs = conv1,
pool_size = 2,
strides = 2)
## Second convolution layer
conv2 = tf.layers.conv1d(inputs = maxp1,
filters = 16,
kernel_size = 7,
padding = 'SAME',
activation = tf.nn.relu)
maxp2 = tf.layers.max_pooling1d(inputs = conv2,
pool_size = 2,
strides = 2)
maxp2_re = tf.contrib.layers.flatten(maxp2)
### Fully connected (= dense connected)
hidden = tf.layers.dense(inputs = maxp2_re,
units = n_hidden,
activation = tf.nn.relu)
output = tf.layers.dense(inputs = hidden,
units = n_output)
return output
In [24]:
# define loss and optimizer
LR = 0.0005
pred = net(x)
loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(labels = y, logits = pred))
optm = tf.train.AdamOptimizer(LR).minimize(loss)
6. Batch Maker¶
In [25]:
n_batch = 64
dataset_train = tf.data.Dataset.from_tensor_slices((X_train, y_train))
dataset_train = dataset_train.shuffle(100000).repeat().batch(n_batch)
iterator_train = dataset_train.make_one_shot_iterator()
next_batch_train = iterator_train.get_next()
dataset_test = tf.data.Dataset.from_tensor_slices((X_test, y_test))
dataset_test = dataset_test.shuffle(100000).repeat().batch(n_batch)
iterator_test = dataset_test.make_one_shot_iterator()
next_batch_test = iterator_test.get_next()
7. Optimize¶
In [26]:
sess = tf.Session()
sess.run(tf.global_variables_initializer())
n_iter = 1500
n_prt = 100
for i in range(n_iter):
train_x, train_y = sess.run(next_batch_train)
sess.run(optm, feed_dict = {x: train_x, y: train_y})
c = sess.run(loss, feed_dict = {x: train_x, y: train_y})
if i % n_prt == 0:
print("Iter: {}".format(i), "Cost: {}".format(c))
8. Test or Evaluation¶
In [27]:
test_pred = sess.run(pred, feed_dict = {x: X_test})
test_pred = 1/(1 + np.exp(-test_pred))
accuracy = np.mean(np.equal(y_test > 0.5, test_pred > 0.5))
print('Accuracy: {0:.2f} %'.format(accuracy * 100))
In [28]:
# confusion matrix
pd.crosstab((y_test[:,0] > 0.5) * 1, (test_pred[:,0] > 0.5) * 1, rownames = ['True'], colnames = ['Predicted'], margins = True)
Out[28]:
In [29]:
def prob_failure(machine_id):
machine_df = test_df[test_df['id'] == machine_id]
machine_test = gen_sequence(machine_df, n_step, x_cols)
test_pred = sess.run(pred, feed_dict = {x: machine_test})
test_pred = 1/(1 + np.exp(-test_pred))
failure_prob = list(test_pred[-1] * 100)[0]
return failure_prob
In [30]:
machine_id = 49
print('Probability that machine will fail within 30 days: {0:.2f} %'.format(prob_failure(machine_id)))